package net.apexes.wsonrpc.demo.server;

import io.netty.handler.codec.http.HttpResponseStatus;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.http.ServerWebSocket;
import net.apexes.wsonrpc.core.WebSocketSession;
import net.apexes.wsonrpc.core.WsonrpcConfig;
import net.apexes.wsonrpc.server.PathAcceptors;
import net.apexes.wsonrpc.server.WsonrpcServer;
import net.apexes.wsonrpc.server.support.VertxWsonrpcHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author <a href="mailto:hedyn@foxmail.com">HeDYn</a>
 */
public class VertxDemoServerImpl extends AbstractDemoWsonrpcServer {

    private static final Logger LOG = LoggerFactory.getLogger(VertxDemoServerImpl.class);
    
    private final Vertx vertx;

    private VertxDemoWsonrpcVerticle verticle;

    public VertxDemoServerImpl() {
        vertx = Vertx.vertx();
    }
    
    @Override
    public WsonrpcServer create(int port, String path, WsonrpcConfig config) {
        VertxWsonrpcHandler wsonrpcHandler = new VertxWsonrpcHandler(config, PathAcceptors.startWithPath(path)) {
            @Override
            protected void onOpen(ServerWebSocket webSocket, WebSocketSession session) {
                LOG.info("connect {} {} ...", session.getId(), webSocket.path());
                Promise<Integer> promise = Promise.promise();
                webSocket.setHandshake(promise.future(), res -> {
                    if (res.succeeded()) {
                        LOG.info("onOpen {} {} ...", session.getId(), webSocket.path());
                        super.onOpen(webSocket, session);
                    }
                });
                vertx.executeBlocking(() -> {
                    Thread.sleep(2000);
                    if (webSocket.path().endsWith("/0")) {
                        promise.tryComplete(HttpResponseStatus.UNAUTHORIZED.code());
                    } else {
                        promise.tryComplete(HttpResponseStatus.SWITCHING_PROTOCOLS.code());
                    }
                    return null;
                }, false);
            }
        };
        wsonrpcHandler.setPingListener((sessionId, bytes) -> LOG.info("onPing: sessionId={}, length={}", sessionId, bytes.length));
        verticle = new VertxDemoWsonrpcVerticle(port, wsonrpcHandler);
        return wsonrpcHandler.getWsonrpcServer();
    }
    
    @Override
    public void startup() throws Exception {
        vertx.deployVerticle(verticle);
    }
    
    @Override
    public void shutdown() throws Exception {
        vertx.close();
    }

    static class VertxDemoWsonrpcVerticle extends AbstractVerticle {

        private final int port;
        private final VertxWsonrpcHandler wsonrpcHandler;

        public VertxDemoWsonrpcVerticle(int port, VertxWsonrpcHandler wsonrpcHandler) {
            this.port = port;
            this.wsonrpcHandler = wsonrpcHandler;
        }

        @Override
        public void start() {
            LOG.debug("start... {}", this);
            HttpServerOptions options = new HttpServerOptions()
                    .setIdleTimeout(60)
                    .setRegisterWebSocketWriteHandlers(true);
            vertx.createHttpServer(options)
                    .webSocketHandler(wsonrpcHandler)
                    .listen(port);
        }

    }
}
